package com.zhang.spark_2.com.zhang.core.req

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @title: 页面单跳转换率统计
 * @author: zhang
 * @date: 2022/2/16 17:16 
 */
object Spark03_req {

  def main(args: Array[String]): Unit = {
    //TODO 获取环境
    val conf = new SparkConf().setMaster("local").setAppName("WordCount")
    val sc = new SparkContext(conf);
    val userVisit: RDD[String] = sc.textFile("data/user_visit_action.txt")

    val mapObjs: RDD[UserVisitAction] = userVisit.map(
      line => {
        val data: Array[String] = line.split("_")
        UserVisitAction(
          data(0),
          data(1).toLong,
          data(2),
          data(3).toLong,
          data(4),
          data(5),
          data(6).toLong,
          data(7).toLong,
          data(8),
          data(9),
          data(10),
          data(11),
          data(12).toLong
        )
      }
    )

    mapObjs.cache()

    //todo 计算分子
    val pageFlow: RDD[((Long, Long), Int)] = mapObjs
      .groupBy(_.session_id)
      .mapValues(
        iter => {
          val actions: List[UserVisitAction] = iter.toList.sortBy(_.action_time)
          val pageIds: List[Long] = actions.map(_.page_id)
          val tuples: List[(Long, Long)] = pageIds.zip(pageIds.tail)
          tuples.map(
            t => {
              ((t._1, t._2), 1)
            }
          )
        }
      )
      .map(_._2)
      .flatMap(list => list)
      .reduceByKey(_ + _)

    //todo 计算分母
    val pageIds: Map[Long, Int] = mapObjs
      .map(
        obj => (obj.page_id, 1)
      )
      .reduceByKey(_ + _)
      .collect()
      .toMap

    //todo 单跳转换率
    pageFlow.foreach {
      case ((p1, p2), cnt) => {
        val count: Int = pageIds.getOrElse(p1, 0)
        println(s"页面$p1 ---> $p2:转化率为：" + (cnt.toDouble * 100 / count) + "%")
      }
    }


    sc.stop()
  }


  //用户访问动作表
  case class UserVisitAction(
      date: String, //用户点击行为的日期
      user_id: Long, //用户的ID
      session_id: String, //Session的ID
      page_id: Long, //某个页面的ID
      action_time: String, //动作的时间点
      search_keyword: String, //用户搜索的关键词
      click_category_id: Long, //某一个商品品类的ID
      click_product_id: Long, //某一个商品的ID
      order_category_ids: String, //一次订单中所有品类的ID集合
      order_product_ids: String, //一次订单中所有商品的ID集合
      pay_category_ids: String, //一次支付中所有品类的ID集合
      pay_product_ids: String, //一次支付中所有商品的ID集合
      city_id: Long //城市 id
)

}
